package PhoneLog.PhoneLog2;




import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class PhoneMapper extends Mapper<LongWritable, Text,Text, FlowBean> {
    //keyIn--->LongWritable：  每行偏移量
    // ValueIn---->Text:每行的数据  1363157985066 	13726230503	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200 	13726230503	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200

    //KeyOut--> Text   电话号码：1363157985066
    //ValueOut--> FlowBean      FlowBean封装好的属性
    //--->private long upFlow;//上行流量   private long downFlow;//下行流量  private long sumFlow;//总流量

    //重写map方法
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1.获取到行数据
        String line = value.toString();
        //2.切割数据的字段
        String[] files = line.split("\t");

        //3.获取数据
        //3.1获取需求数据（手机号-->keyout）
        String phoneNum = files[1];
        //3.2获取上下行流量
        Long upFlow = Long.parseLong(files[files.length - 3]);//上行流量
        Long downFlow = Long.parseLong(files[files.length - 2]);//下行流量
        //4.封装数据(因为获取需求数据的类型要和Mapper中keyout与valueout的类型一致）
           //4.1封装key值（手机号）
        Text k = new Text();
        k.set(phoneNum);
           //4.2封装FlowBean
        FlowBean v = new FlowBean();
        v.setDownFlow(downFlow);//下行流量
        v.setUpFlow(upFlow);//上行流量
        //v.setSumFlow(downFlow+upFlow);//总流量

        //5.写出数据
        context.write(k,v);
    }

}
